package cn.lianxi.kaoshi._12_;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Driver {

	/**
	 * 如何加入黑名单
	 * 
	 * 1. 加入黑名单 --> 获取
	 * 
	 * 2. 黑名单数据 --> SET
	 * 
	 * 3. 利用SET 对 key做过滤 --> mapper
	 * 
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {

		// 1. 拿到集群的配置信息
		Configuration conf = new Configuration(); // 补充: 可以往容器中设置自定义的参数
		// 设置黑名单
		conf.setStrings(
				"blackInfo",
				readHdfsFile2Array("/user/zhaoyingfei/tmp/black.txt", "utf-8",
						conf));

		// 2. 创建job
		Job job = Job.getInstance();
		// 3. 设置 JarClass
		job.setJarByClass(Driver.class);
		// 4. 设置Mapper Reducer
		job.setMapperClass(TokenizerMapper.class);
		job.setReducerClass(IntSumReducer.class);
		// 5. 输入输出的类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		// 6. 设置输入和输出的路径
		FileInputFormat.addInputPath(job, new Path("/tmp/zyf/input/wordcount"));
		FileOutputFormat.setOutputPath(job, new Path("/tmp/zyf/out5"));

		// 设置reduce_nums
		job.setNumReduceTasks(1);

		// 7. 启动执行
		System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

	private static String[] readHdfsFile2Array(String hdfsPath,
			String charsetName, Configuration conf) {
		// 1. 判断合法性
		// FileUtil.checkPath(hdfsPath);

		// 存放数据
		List<String> result = new ArrayList<String>();

		BufferedReader br = null;
		FileSystem fs = null;
		try {
			// 获取文件系统
			fs = FileSystem.get(conf);
			// 打开文件流
			FSDataInputStream in = fs.open(new Path(hdfsPath));
			InputStreamReader isr = new InputStreamReader(in, charsetName);
			br = new BufferedReader(isr);
			// 读数据
			String line = null;
			while ((line = br.readLine()) != null) {
				result.add(line.trim());
			}

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				br.close();
				fs.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		// set --> array
		String[] arr = new String[result.size()];
		for (int i = 0; i < arr.length; i++) {
			arr[i] = result.get(i);
		}

		return arr;
	}

}
